package com.etlmaster.datax.plugins.reader.rationaldbreader;

import java.util.ArrayList;
import java.util.List;

import com.dp.nebula.wormhole.common.AbstractSplitter;
import com.dp.nebula.wormhole.common.interfaces.IParam;
import com.etlmaster.executor.utils.DataCollectionUtils;

public class RationalDbReaderKeySplitter extends AbstractSplitter {

	private String tableName;

	private String columns;

	private String where;

	private int blockSize;

	private String url;

	private String port = "3306";

	private String dbname;

	private int concurrency;

	private boolean needSplit;

	private String sql;

	private String dbType;
	private String splitSql;
	private RationalDbReaderSplitter candidateSplitter = new RationalDbReaderSplitter();

	private static final int DEFAULT_BLOCK_SIZE = 100000;

	@Override
	public void init(IParam jobParams) {
		super.init(jobParams);
		param.getValue(ParamKey.autoIncKey, "");
		tableName = param.getValue(ParamKey.tableName, "");
		columns = param.getValue(ParamKey.columns, "");
		where = param.getValue(ParamKey.where, "");
		blockSize = param.getIntValue(ParamKey.blockSize, DEFAULT_BLOCK_SIZE);
		url = param.getValue(ParamKey.url, "");
		port = param.getValue(ParamKey.port, this.port);
		dbname = param.getValue(ParamKey.dbname, "");
		sql = param.getValue(ParamKey.sql, "");
		concurrency = param.getIntValue(ParamKey.concurrency, 1);
		needSplit = param.getBooleanValue(ParamKey.needSplit, true);
		splitSql = param.getValue(ParamKey.splitSql, " limit %s and %d");
		dbType=DataCollectionUtils.getDatasourceType(param.getValue(ParamKey.driverClass));
		candidateSplitter.init(jobParams);
	}

	@Override
	public List<IParam> split() {
		List<IParam> paramList = new ArrayList<IParam>();
		if (needSplit) {
			
			if (!sql.isEmpty() ) {
				String SplitSql = "";
				long size = param.getLongValue(
						RationalDbReaderPeriphery.DATA_AMOUNT_KEY, -1);
				for (long i = 0; i <= size / blockSize; i++) 
				{
					IParam paramSplitted = param.clone();
					SplitSql = getSplitSqlByDsType(i);
					System.out.println(SplitSql);
					paramSplitted.putValue(ParamKey.sql, SplitSql);
					paramList.add(paramSplitted);
				}
				
				return paramList;
			}

		}else{
			paramList.add(param);
		}
		return paramList;
	}
	public String getSplitSqlByDsType(long i){
		String rtnString="";
		if(dbType.equalsIgnoreCase("oracle")){
			rtnString=String.format(splitSql, (i+1)*blockSize,(i)*blockSize);
		}else{
			rtnString=String.format(sql+splitSql, blockSize,i*blockSize);;
		}
		return rtnString;
	}

}
